package com.tencent.sr.iris.activity.service.kafka;

import com.alibaba.fastjson.JSON;
import com.dianping.cat.Cat;
import com.tencent.sr.iris.activity.common.exception.BizCommErrorCode;
import com.tencent.sr.rmall.springbootstarter.kafka.KafkaMetaMessage;
import com.tencent.sr.rmall.springbootstarter.monitor.util.MonitorUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.ProducerListener;
import org.springframework.lang.Nullable;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * 描述用途
 *
 * @author ylongzhang
 * @date 2022/07/18
 */
@Slf4j
@Component
public class KafkaMsgSender {

    @Resource
    private KafkaTemplate<String, KafkaMetaMessage> kafkaTemplate;

    /**
     * 发送消息
     *
     * @param topic kafka主题
     * @param messageKey 分区key。用来保证局部有序
     * @param data 待发送数据
     */
    public void send(String topic, @Nullable String messageKey, String data) {
        KafkaMetaMessage msg = new KafkaMetaMessage();
        msg.setData(data);
        kafkaTemplate.setProducerListener(new ProducerListener<String, KafkaMetaMessage>() {
            @Override
            public void onSuccess(String topic, Integer partition, String key,
                    KafkaMetaMessage value, RecordMetadata recordMetadata) {
                String body = JSON.toJSONString(value);

                Cat.logEvent("kafka" + "_" + topic, body);
                log.info("kafkaMqMessageSender success:{}:{}", topic, body);
            }

            @Override
            public void onError(String topic, Integer partition, String key,
                    KafkaMetaMessage value, Exception exception) {
                String body = JSON.toJSONString(value);
                log.error(MonitorUtil.getErrorMsg(BizCommErrorCode.OUT_LET_KAFKA_TOPIC_ERROR.getCode(),
                        "MessageSendError : " + body));
            }
        });
        kafkaTemplate.send(topic, messageKey, msg);
    }
}
